use anyhow::Result;
use std::sync::Arc;
use std::net::SocketAddr;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
use tracing::{info, debug, error};

use crate::config::ServerConfig;
use crate::database::Database;
use crate::ftp::commands::CommandProcessor;
use crate::services::AuthManager;
use crate::filesystem::FileSystemManager;
use crate::ftp::data_transfer::{DataTransferManager, TransferMode as DataTransferMode, TransferType, TransferStats};
use shared::ftp_protocol::FtpCommand;
use shared::types::{FtpResponse, FtpResponseCode, TransferMode, Session as DbSession};

pub struct FtpSession {
    stream: BufReader<TcpStream>,
    peer_addr: SocketAddr,
    database: Arc<Database>,
    config: ServerConfig,
    auth_manager: Arc<AuthManager>,
    is_authenticated: bool,
    current_user_id: Option<i64>,
    current_directory: String,
    transfer_mode: TransferMode,
    data_port: Option<u16>,
    passive_listener: Option<tokio::net::TcpListener>,
    resume_offset: Option<u64>,
    file_system_manager: FileSystemManager,
    data_transfer_manager: DataTransferManager,
    pending_username: Option<String>,
    session_data: Option<DbSession>,
    // 新增字段
    transfer_type: TransferType,
    data_transfer_mode: DataTransferMode,
    transfer_stats: Option<TransferStats>,
}

impl FtpSession {
    pub async fn new(
        stream: TcpStream,
        peer_addr: SocketAddr,
        database: Arc<Database>,
        config: ServerConfig,
        auth_manager: AuthManager,
    ) -> Result<Self> {
        let stream = BufReader::new(stream);
        let base_path = std::path::PathBuf::from(&config.filesystem.root_directory);
        let auth_manager = Arc::new(auth_manager);

        Ok(Self {
            stream,
            peer_addr,
            database,
            config,
            auth_manager,
            is_authenticated: false,
            current_user_id: None,
            current_directory: "/".to_string(),
            transfer_mode: TransferMode::Passive,
            data_port: None,
            passive_listener: None,
            resume_offset: None,
            file_system_manager: FileSystemManager::new(base_path),
            data_transfer_manager: DataTransferManager::new(),
            pending_username: None,
            session_data: None,
            // 初始化新字段
            transfer_type: TransferType::Binary,
            data_transfer_mode: DataTransferMode::Passive,
            transfer_stats: None,
        })
    }
    
    pub async fn run(&mut self) -> Result<()> {
        let mut line = String::new();
        
        loop {
            line.clear();
            
            match self.stream.read_line(&mut line).await {
                Ok(0) => {
                    info!("Client {} disconnected", self.peer_addr);
                    break;
                }
                    Ok(_) => {
                        let command_str = line.trim();
                        debug!("Received command: {}", command_str);
                        
                        if let Some(command) = FtpCommand::from_string(command_str) {
                            // 使用命令处理器处理命令
                            let response = CommandProcessor::process_command(&command, self).await;
                            self.send_response(response).await?;
                            
                            // 检查是否需要断开连接
                            if matches!(command, FtpCommand::Quit) {
                                break;
                            }
                        } else {
                            let response = FtpResponse::new(
                                FtpResponseCode::SyntaxError,
                                "Invalid command"
                            );
                            self.send_response(response).await?;
                        }
                    }
                Err(e) => {
                    error!("Error reading from client {}: {}", self.peer_addr, e);
                    break;
                }
            }
        }
        
        Ok(())
    }
    
    // 权限检查方法
    pub async fn check_permission(&self, permission: shared::types::PermissionType, path: &str) -> Result<bool> {
        if let Some(user_id) = self.current_user_id {
            self.auth_manager.check_permission(user_id, permission, path).await
        } else {
            Ok(false)
        }
    }
    
    // 目录操作命令实现
    async fn handle_cwd_command(&mut self, path: &str) -> FtpResponse {
        if path.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Path required");
        }

        // 构建目标路径
        let new_path = if path == ".." {
            "/".to_string() // 简化处理，回到根目录
        } else if path.starts_with('/') {
            path.to_string()
        } else {
            format!("{}/{}", self.current_directory, path)
        };

        // 检查读取权限
        match self.check_permission(shared::types::PermissionType::Read, &new_path).await {
            Ok(true) => {
                self.current_directory = new_path;
                FtpResponse::new(FtpResponseCode::RequestedFileActionOk, "Directory changed")
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }

    async fn handle_cdup_command(&mut self) -> FtpResponse {
        if self.current_directory == "/" {
            FtpResponse::new(FtpResponseCode::RequestedFileActionOk, "Already at root directory")
        } else {
            // 检查父目录的读取权限
            match self.check_permission(shared::types::PermissionType::Read, "/").await {
                Ok(true) => {
                    self.current_directory = "/".to_string();
                    FtpResponse::new(FtpResponseCode::RequestedFileActionOk, "Directory changed")
                }
                Ok(false) => {
                    FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
                }
                Err(_) => {
                    FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
                }
            }
        }
    }

    async fn handle_mkd_command(&mut self, path: &str) -> FtpResponse {
        if path.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Path required");
        }

        let full_path = if path.starts_with('/') {
            path.to_string()
        } else {
            format!("{}/{}", self.current_directory, path)
        };

        // 检查写入权限
        match self.check_permission(shared::types::PermissionType::Write, &full_path).await {
            Ok(true) => {
                // 简化实现，实际应该创建目录
                FtpResponse::new(FtpResponseCode::PathnameCreated, &format!("\"{}\" directory created", path))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }

    async fn handle_rmd_command(&mut self, path: &str) -> FtpResponse {
        if path.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Path required");
        }

        let full_path = if path.starts_with('/') {
            path.to_string()
        } else {
            format!("{}/{}", self.current_directory, path)
        };

        // 检查删除权限
        match self.check_permission(shared::types::PermissionType::Delete, &full_path).await {
            Ok(true) => {
                // 简化实现，实际应该删除目录
                FtpResponse::new(FtpResponseCode::RequestedFileActionOk, &format!("\"{}\" directory removed", path))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    // 文件操作命令实现
    async fn handle_list_command(&mut self, path: Option<&str>) -> FtpResponse {
        let target_path = path.unwrap_or(&self.current_directory);
        
        // 检查读取权限
        match self.check_permission(shared::types::PermissionType::Read, target_path).await {
            Ok(true) => {
                // 检查数据连接是否已建立
                if !self.is_data_connection_ready() {
                    return FtpResponse::new(FtpResponseCode::DataConnectionNotOpen, "Use PASV first");
                }
                
                // 发送数据传输开始响应
                FtpResponse::new(FtpResponseCode::DataConnectionAlreadyOpen, "Opening ASCII mode data connection for file list")
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_nlst_command(&mut self, path: Option<&str>) -> FtpResponse {
        let target_path = path.unwrap_or(&self.current_directory);
        
        // 检查读取权限
        match self.check_permission(shared::types::PermissionType::Read, target_path).await {
            Ok(true) => {
                // 检查数据连接是否已建立
                if !self.is_data_connection_ready() {
                    return FtpResponse::new(FtpResponseCode::DataConnectionNotOpen, "Use PASV first");
                }
                
                // 发送数据传输开始响应
                FtpResponse::new(FtpResponseCode::DataConnectionAlreadyOpen, "Opening ASCII mode data connection for name list")
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_retr_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        let file_path = self.build_file_path(filename);
        
        // 检查读取权限
        match self.check_permission(shared::types::PermissionType::Read, &file_path).await {
            Ok(true) => {
                // 检查数据连接是否已建立
                if !self.is_data_connection_ready() {
                    return FtpResponse::new(FtpResponseCode::DataConnectionNotOpen, "Use PASV first");
                }
                
                // 发送数据传输开始响应
                FtpResponse::new(FtpResponseCode::DataConnectionAlreadyOpen, &format!("Opening ASCII mode data connection for {}", filename))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_stor_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        let file_path = self.build_file_path(filename);
        
        // 检查写入权限
        match self.check_permission(shared::types::PermissionType::Write, &file_path).await {
            Ok(true) => {
                // 检查数据连接是否已建立
                if !self.is_data_connection_ready() {
                    return FtpResponse::new(FtpResponseCode::DataConnectionNotOpen, "Use PASV first");
                }
                
                // 发送数据传输开始响应
                FtpResponse::new(FtpResponseCode::DataConnectionAlreadyOpen, &format!("Opening ASCII mode data connection for {}", filename))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_appe_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        let file_path = self.build_file_path(filename);
        
        // 检查写入权限
        match self.check_permission(shared::types::PermissionType::Write, &file_path).await {
            Ok(true) => {
                // 检查数据连接是否已建立
                if !self.is_data_connection_ready() {
                    return FtpResponse::new(FtpResponseCode::DataConnectionNotOpen, "Use PASV first");
                }
                
                // 发送数据传输开始响应
                FtpResponse::new(FtpResponseCode::DataConnectionAlreadyOpen, &format!("Opening ASCII mode data connection for {}", filename))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_dele_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        let file_path = self.build_file_path(filename);
        
        // 检查删除权限
        match self.check_permission(shared::types::PermissionType::Delete, &file_path).await {
            Ok(true) => {
                // 简化实现，实际应该删除文件
                FtpResponse::new(FtpResponseCode::RequestedFileActionOk, &format!("File {} deleted", filename))
            }
            Ok(false) => {
                FtpResponse::new(FtpResponseCode::FileNotFound, "Permission denied")
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::LocalError, "Permission check failed")
            }
        }
    }
    
    async fn handle_size_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        // 简化实现，实际应该返回文件大小
        FtpResponse::new(FtpResponseCode::FileStatusOk, "12345") // 示例文件大小
    }
    
    async fn handle_mdtm_command(&mut self, filename: &str) -> FtpResponse {
        if filename.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Filename required");
        }
        
        // 简化实现，实际应该返回文件修改时间
        FtpResponse::new(FtpResponseCode::FileStatusOk, "20240101120000") // 示例时间戳
    }
    
    // 传输模式命令实现
    async fn handle_pasv_command(&mut self) -> FtpResponse {
        // 创建被动模式监听器
        match self.create_passive_listener().await {
            Ok(port) => {
                self.transfer_mode = TransferMode::Passive;
                // 返回PASV响应，格式: Entering Passive Mode (h1,h2,h3,h4,p1,p2)
                let local_ip = "127.0.0.1"; // 简化处理，实际应该获取服务器IP
                let ip_parts: Vec<&str> = local_ip.split('.').collect();
                let p1 = (port >> 8) as u8;
                let p2 = (port & 0xFF) as u8;
                
                let pasv_response = format!(
                    "Entering Passive Mode ({},{},{},{},{},{})",
                    ip_parts[0], ip_parts[1], ip_parts[2], ip_parts[3], p1, p2
                );
                
                FtpResponse::new(FtpResponseCode::EnteringPassiveMode, &pasv_response)
            }
            Err(_) => {
                FtpResponse::new(FtpResponseCode::ServiceNotAvailable, "Failed to create passive connection")
            }
        }
    }
    
    
    async fn handle_port_command(&mut self, addr: &str) -> FtpResponse {
        if addr.is_empty() {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Address required");
        }
        
        // 解析PORT命令的地址格式: h1,h2,h3,h4,p1,p2
        let parts: Vec<&str> = addr.split(',').collect();
        if parts.len() != 6 {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Invalid address format");
        }
        
        // 解析IP地址
        let _ip_str = format!("{}.{}.{}.{}", parts[0], parts[1], parts[2], parts[3]);
        let port = (parts[4].parse::<u16>().unwrap_or(0) << 8) + parts[5].parse::<u16>().unwrap_or(0);
        
        if port == 0 {
            return FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Invalid port number");
        }
        
        // 存储客户端数据连接信息
        self.transfer_mode = TransferMode::Active;
        self.data_port = Some(port);
        
        FtpResponse::new(FtpResponseCode::CommandOk, "PORT command successful")
    }
    
    async fn handle_type_command(&mut self, mode: &str) -> FtpResponse {
        match mode.to_uppercase().as_str() {
            "A" | "ASCII" => {
                FtpResponse::new(FtpResponseCode::CommandOk, "Type set to A")
            }
            "I" | "IMAGE" | "BINARY" => {
                FtpResponse::new(FtpResponseCode::CommandOk, "Type set to I")
            }
            _ => {
                FtpResponse::new(FtpResponseCode::ParameterSyntaxError, "Invalid type")
            }
        }
    }
    
    async fn handle_rest_command(&mut self, offset: u64) -> FtpResponse {
        self.resume_offset = Some(offset);
        FtpResponse::new(FtpResponseCode::RequestedFileActionOk, &format!("Restarting at {}", offset))
    }
    
    
    async fn accept_data_connection(&mut self) -> Result<tokio::net::TcpStream> {
        match self.transfer_mode {
            TransferMode::Passive => {
                if let Some(listener) = self.passive_listener.take() {
                    let (stream, _) = listener.accept().await?;
                    Ok(stream)
                } else {
                    Err(anyhow::anyhow!("No passive listener available"))
                }
            }
            TransferMode::Active => {
                if let Some(port) = self.data_port {
                    // 构建客户端数据连接地址
                    let client_ip = self.peer_addr.ip();
                    let data_addr = std::net::SocketAddr::new(client_ip, port);
                    let stream = tokio::net::TcpStream::connect(data_addr).await?;
                    Ok(stream)
                } else {
                    Err(anyhow::anyhow!("No data port configured"))
                }
            }
        }
    }
    
    async fn transfer_directory_list(&mut self, path: &str) -> Result<()> {
        let data_stream = self.accept_data_connection().await?;
        let mut data_stream = tokio::io::BufWriter::new(data_stream);
        
        // 使用DataTransferManager传输目录列表
        let directory_path = std::path::PathBuf::from(path);
        let bytes_transferred = self.data_transfer_manager.transfer_directory_list(&mut data_stream, &directory_path).await?;
        
        // 发送传输完成响应
        let response = FtpResponse::new(FtpResponseCode::ClosingDataConnection, &format!("Transfer complete. {} bytes transferred", bytes_transferred));
        self.send_response(response).await?;
        
        Ok(())
    }
    
    async fn transfer_file_download(&mut self, file_path: &str) -> Result<()> {
        let data_stream = self.accept_data_connection().await?;
        let mut data_stream = tokio::io::BufWriter::new(data_stream);
        
        // 使用DataTransferManager下载文件
        let path = std::path::PathBuf::from(file_path);
        let bytes_transferred = self.data_transfer_manager.download_file(&mut data_stream, &path, self.resume_offset, self.transfer_type).await?;
        
        // 发送传输完成响应
        let response = FtpResponse::new(FtpResponseCode::ClosingDataConnection, &format!("Transfer complete. {} bytes transferred", bytes_transferred));
        self.send_response(response).await?;
        
        // 清除断点续传偏移量
        self.resume_offset = None;
        
        Ok(())
    }
    
    async fn transfer_file_upload(&mut self, file_path: &str) -> Result<()> {
        let data_stream = self.accept_data_connection().await?;
        let mut data_stream = tokio::io::BufReader::new(data_stream);
        
        // 使用DataTransferManager上传文件
        let path = std::path::PathBuf::from(file_path);
        let bytes_transferred = self.data_transfer_manager.upload_file(&mut data_stream, &path, self.resume_offset, self.transfer_type).await?;
        
        // 发送传输完成响应
        let response = FtpResponse::new(FtpResponseCode::ClosingDataConnection, &format!("Transfer complete. {} bytes transferred", bytes_transferred));
        self.send_response(response).await?;
        
        // 清除断点续传偏移量
        self.resume_offset = None;
        
        Ok(())
    }
    
    pub async fn send_response(&mut self, response: FtpResponse) -> Result<()> {
        let response_str = format!("{} {}\r\n", response.code as u16, response.message);
        debug!("Sending response: {}", response_str.trim());
        
        self.stream.get_mut().write_all(response_str.as_bytes()).await?;
        self.stream.get_mut().flush().await?;
        
        Ok(())
    }
    
    // Getters for command processor
    pub fn is_authenticated(&self) -> bool {
        self.is_authenticated
    }
    
    pub fn current_user_id(&self) -> Option<i64> {
        self.current_user_id
    }
    
    pub fn current_directory(&self) -> &str {
        &self.current_directory
    }
    
    pub fn peer_addr(&self) -> SocketAddr {
        self.peer_addr
    }
    
    pub fn database(&self) -> &Arc<Database> {
        &self.database
    }
    
    pub fn config(&self) -> &ServerConfig {
        &self.config
    }
    
    pub fn auth_manager(&self) -> &Arc<AuthManager> {
        &self.auth_manager
    }
    
    pub fn get_pending_username(&self) -> Option<&String> {
        self.pending_username.as_ref()
    }
    
    pub fn session_data(&self) -> Option<&DbSession> {
        self.session_data.as_ref()
    }
    
    // Setters for command processor
    pub fn set_authenticated(&mut self, session_data: DbSession) {
        self.is_authenticated = true;
        self.current_user_id = Some(session_data.user_id);
        self.session_data = Some(session_data);
        self.current_directory = "/".to_string(); // 重置到用户根目录
    }
    
    pub fn set_current_user_id(&mut self, user_id: Option<i64>) {
        self.current_user_id = user_id;
    }
    
    pub fn set_current_directory(&mut self, directory: String) {
        self.current_directory = directory;
    }
    
    pub fn set_pending_username(&mut self, username: String) {
        self.pending_username = Some(username);
    }
    
    pub fn clear_pending_username(&mut self) {
        self.pending_username = None;
    }
    
    // 数据传输相关方法
    pub fn is_data_connection_ready(&self) -> bool {
        match self.transfer_mode {
            TransferMode::Passive => self.passive_listener.is_some(),
            TransferMode::Active => self.data_port.is_some(),
        }
    }
    
    pub fn build_file_path(&self, filename: &str) -> String {
        if filename.starts_with('/') {
            filename.to_string()
        } else {
            format!("{}/{}", self.current_directory, filename)
        }
    }
    
    pub async fn create_passive_listener(&mut self) -> Result<u16> {
        // 创建监听器并绑定到随机端口
        let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await?;
        let local_addr = listener.local_addr()?;
        let port = local_addr.port();
        
        // 存储监听器以供后续使用
        self.passive_listener = Some(listener);
        self.data_port = Some(port);
        
        Ok(port)
    }
    
    pub fn set_transfer_mode(&mut self, mode: TransferMode) {
        self.transfer_mode = mode;
    }
    
    pub fn set_data_port(&mut self, port: Option<u16>) {
        self.data_port = port;
    }
    
    pub fn set_resume_offset(&mut self, offset: Option<u64>) {
        self.resume_offset = offset;
    }

    // 新增的数据传输方法
    pub fn set_transfer_type(&mut self, transfer_type: TransferType) {
        self.transfer_type = transfer_type;
    }

    pub fn get_transfer_type(&self) -> TransferType {
        self.transfer_type
    }

    pub fn set_data_transfer_mode(&mut self, mode: DataTransferMode) {
        self.data_transfer_mode = mode;
    }

    pub fn get_data_transfer_mode(&self) -> DataTransferMode {
        self.data_transfer_mode
    }

    pub fn get_transfer_stats(&self) -> Option<&TransferStats> {
        self.transfer_stats.as_ref()
    }

    pub fn set_transfer_stats(&mut self, stats: TransferStats) {
        self.transfer_stats = Some(stats);
    }

    /// 执行文件上传
    pub async fn execute_file_upload(&mut self, filename: &str) -> Result<TransferStats> {
        let file_path = self.build_file_path(filename);
        let path_buf = std::path::PathBuf::from(&file_path);

        // 创建数据连接
        let data_stream = self.create_data_connection().await?;
        let mut data_reader = tokio::io::BufReader::new(data_stream);

        // 执行上传
        let stats = self.data_transfer_manager.upload_file(
            &mut data_reader,
            &path_buf,
            self.resume_offset,
            self.transfer_type,
        ).await?;

        self.set_transfer_stats(stats.clone());
        Ok(stats)
    }

    /// 执行文件下载
    pub async fn execute_file_download(&mut self, filename: &str) -> Result<TransferStats> {
        let file_path = self.build_file_path(filename);
        let path_buf = std::path::PathBuf::from(&file_path);

        // 创建数据连接
        let data_stream = self.create_data_connection().await?;
        let mut data_writer = tokio::io::BufWriter::new(data_stream);

        // 执行下载
        let stats = self.data_transfer_manager.download_file(
            &mut data_writer,
            &path_buf,
            self.resume_offset,
            self.transfer_type,
        ).await?;

        self.set_transfer_stats(stats.clone());
        Ok(stats)
    }

    /// 执行目录列表传输
    pub async fn execute_directory_list(&mut self, path: Option<&str>) -> Result<TransferStats> {
        let list_path = if let Some(p) = path {
            self.build_file_path(p)
        } else {
            self.current_directory.clone()
        };

        let path_buf = std::path::PathBuf::from(&list_path);

        // 创建数据连接
        let data_stream = self.create_data_connection().await?;
        let mut data_writer = tokio::io::BufWriter::new(data_stream);

        // 执行目录列表传输
        let stats = self.data_transfer_manager.transfer_directory_list(
            &mut data_writer,
            &path_buf,
        ).await?;

        self.set_transfer_stats(stats.clone());
        Ok(stats)
    }

    /// 创建数据连接
    async fn create_data_connection(&mut self) -> Result<tokio::net::TcpStream> {
        match self.data_transfer_mode {
            DataTransferMode::Passive => {
                if let Some(listener) = self.passive_listener.take() {
                    let (stream, _) = listener.accept().await?;
                    Ok(stream)
                } else {
                    Err(anyhow::anyhow!("No passive listener available"))
                }
            }
            DataTransferMode::Active => {
                if let Some(port) = self.data_port {
                    let data_addr = std::net::SocketAddr::new(self.peer_addr.ip(), port);
                    let stream = tokio::net::TcpStream::connect(data_addr).await?;
                    Ok(stream)
                } else {
                    Err(anyhow::anyhow!("No data port specified"))
                }
            }
        }
    }

    /// 格式化传输统计信息
    pub fn format_transfer_stats(&self) -> String {
        if let Some(stats) = &self.transfer_stats {
            let speed_str = self.data_transfer_manager.format_speed(stats.speed_bytes_per_sec);
            let size_str = self.data_transfer_manager.format_size(stats.bytes_transferred);
            let duration_str = format!("{:.2}s", stats.duration.as_secs_f64());
            
            format!(
                "Transfer completed: {} in {}, speed: {}",
                size_str, duration_str, speed_str
            )
        } else {
            "No transfer statistics available".to_string()
        }
    }
}
